package day02;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Calendar;
import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * @Description: 自定义数据源
 * @Author: ZYX
 * @Date: 2022/2/9 10:33
 * @Version: 1.0
 */
public class Demo01 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        executionEnvironment.setParallelism(1);

        DataStreamSource<Event> eventDataStreamSource = executionEnvironment.addSource(new ClickSource());

        eventDataStreamSource.print();

        executionEnvironment.execute();
    }

    public static class ClickSource implements SourceFunction<Event>{
        private boolean running = true;
        private String[] userArr = {"Alen","Bob","Carter","Dave","Eden","Frank","Gucci"};
        private String[] urlArr = {"./home","./cart","./fav","./user/1","./product/2"};
        private Random random = new Random();

        @Override
        public void run(SourceContext<Event> sourceContext) throws Exception {
            while (running){
                // collect 方法，向下游发送数据
                sourceContext.collect(
                        new Event(userArr[random.nextInt(userArr.length)],
                                urlArr[random.nextInt(urlArr.length)],
                                Calendar.getInstance().getTimeInMillis())
                );
                TimeUnit.SECONDS.sleep(1L);
            }
        }

        @Override
        public void cancel() {
            running = false;

        }
    }

    public static class Event{
        public String user;
        public String url;
        public Long timeStamp;

        public Event() {
        }

        public Event(String user, String url, Long timeStamp) {
            this.user = user;
            this.url = url;
            this.timeStamp = timeStamp;
        }

        @Override
        public String toString() {
            return "Event{" +
                    "user='" + user + '\'' +
                    ", url='" + url + '\'' +
                    ", timeStamp=" + timeStamp +
                    '}';
        }
    }

}
